package com.lvy.hczn.front.business.config;

import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.lvy.hczn.front.business.service.ConsumeService;
import com.lvy.hczn.front.common.constant.Constants;
import com.lvy.hczn.front.common.constant.MqConstants;
import com.lvy.hczn.front.common.constant.RedisConstants;
import com.lvy.hczn.front.common.core.domain.BaseModel;
import com.lvy.hczn.front.common.core.redis.RedisCache;
import com.lvy.hczn.front.common.exception.UtilException;
import com.lvy.hczn.front.system.domain.business.Warehouse;
import com.lvy.hczn.front.system.service.WarehouseService;
import com.lvy.hczn.mq.rocket.MqProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

@Configuration
@EnableConfigurationProperties(MqProperties.class)
@ConditionalOnProperty(prefix = "rocketmq.lvy", value = "nameSrvAddr")
@Slf4j
public class MqConsumerConfig {

    @Autowired
    private MqProperties mqProperties;
    @Autowired
    private ConsumeService consumeService;

    @Autowired
    private WarehouseService warehouseService;

    @Autowired
    private RedisCache redisCache;

    /**
     * 初始化rocketmq消息监听方式的消费者
     */
    @Bean
    @ConditionalOnProperty(prefix = "rocketmq.lvy", value = "consumerInstanceName")
    public DefaultMQPushConsumer pushConsumer() throws MQClientException {
        try {
            Warehouse warehouse = redisCache.getCacheObject(RedisConstants.WAREHOUSE_WMS);
            if (warehouse == null) {
                throw new UtilException("仓库信息为空");
            }
            if (StrUtil.isEmpty(warehouse.getAreaCode())) {
                throw new UtilException("仓库运营区域为空");
            }
            String suffix = "_" + warehouse.getWarehouseCode();
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqProperties.getConsumerGroupName() + suffix, new AclClientRPCHook(new SessionCredentials(mqProperties.getAclAccessKey(), mqProperties.getAclSecretKey())), new AllocateMessageQueueAveragely());
//            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqProperties.getConsumerGroupName());
            consumer.setNamesrvAddr(mqProperties.getNameSrvAddr());
            consumer.setInstanceName(mqProperties.getConsumerInstanceName() + suffix);

            // 设置批量消费，以提升消费吞吐量，默认是1
            int num = mqProperties.getConsumerBatchMaxSize() == 0 ? 1 : mqProperties.getConsumerBatchMaxSize();
            consumer.setConsumeMessageBatchMaxSize(num);
            //单队列并行消费最大跨度，用于流控
            consumer.setConsumeConcurrentlyMaxSpan(2000);
            // 一个queue最大消费的消息个数，用于流控
            consumer.setPullThresholdForQueue(1000);
            //消息拉取时间间隔，默认为0，即拉完一次立马拉第二次，单位毫秒
            consumer.setPullInterval(1000);
            if (mqProperties.isConsumerBroadcasting()) {
                consumer.setMessageModel(MessageModel.BROADCASTING);
            } else {
                //消费模式，集群消费
                consumer.setMessageModel(MessageModel.CLUSTERING);
            }

            /**
             * 订阅指定topic下tags
             */
            List<String> subscribeList = mqProperties.getSubscribe();
            //订阅运营区域
            //区域topic前缀,一个区域为一个topic，区域下的总仓和前置仓监听此topic,tag以每个仓库编号+后缀区分
            subscribeList.add(MqConstants.TOPIC_WAREHOUSE_AREA + "_" + warehouse.getAreaCode().replace(".", ""));
            for (String subscribe : subscribeList) {
                String[] sub = subscribe.split(":");
                if (sub.length > 1) {
                    consumer.subscribe(sub[0], sub[1]);
                } else {
                    consumer.subscribe(sub[0], "*");
                }
            }
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
                //log.info("消息消费：messages:" + messages);
                try {
                    for (MessageExt message : messages) {
                        consumeService.doBusiness(message);
                        //RECONSUME_LATER：如果业务逻辑消费异常返回如：数据库异常，余额不足扣款失败等一切业务认为消息需要重试的场景(可发邮件或者其他方式提醒数据异常)
                        //对于并发的消费监听器，你可以返回 RECONSUME_LATER 来通知消费者现在不能消费这条消息，并且希望可以稍后重新消费它。
                        // 然后，你可以继续消费其他消息。对于有序的消息监听器，因为你关心它的顺序，所以不能跳过消息，但是你可以返回SUSPEND_CURRENT_QUEUE_A_MOMENT 告诉消费者等待片刻。
                        // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("业务处理异常：" + e.getMessage(), e.getCause());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            return consumer;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("消费异常：" + e.getMessage());
        }
        return null;
    }
}
